package com.panlm.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.*;

public class CustomConsumer {

    public static String KAFKA_CLUSTER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";
    private KafkaConsumer<String, String> consumer;
    private static Map<TopicPartition, Long> currentOffset = new HashMap<>();

    @Before
    public void before() {
        Properties properties = new Properties();
        //Kafka集群
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CLUSTER);
        //消费者组，只要group.id相同，就属于同一个消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "panlm");
        //是否开启自动提交offset功能
        //properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //自动提交offset的时间间隔
        //properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //反序列化Key和Value
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        consumer = new KafkaConsumer<>(properties);
    }

    @After
    public void after() {
        consumer.close();
    }

    //自动提交offset
    @Test
    public void testAutoCommitOffset() {
        consumer.subscribe(Arrays.asList("bigdata"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }

    //手动提交offset: 同步提交
    @Test
    public void testCommitSyncOffset() {
        consumer.subscribe(Arrays.asList("bigdata"));
        while (true) {
            //消费者拉取数据
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset=%d, key=%s, value=%s%n", record.offset(), record.key(), record.value());
            }
            //同步提交，当前线程会阻塞直到offset提交成功
            consumer.commitSync();
        }
    }

    /**
     * 手动提交offset:异步提交
     * 虽然同步提交offset更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞
     * 吐量会收到很大的影响。因此更多的情况下，会选用异步提交offset的方式
     */
    @Test
    public void testCommitAsyncOffset() {
        consumer.subscribe(Arrays.asList("bigdata"));
        while (true) {
            //消费者拉取数据
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset=%d, key=%s, value=%s%n", record.offset(), record.key(), record.value());
            }
            //异步提交
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                    if (e != null) {
                        System.out.println("Commit failed for" + map);
                    }
                }
            });
        }
    }

    //自定义存储offset
    @Test
    public void customSaveOffset() {
        //消费者订阅主题
        consumer.subscribe(Arrays.asList("bigdata"), new ConsumerRebalanceListener() {
            //该方法会在Rebalance之前调用
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                commitOffset(currentOffset);
            }

            //该方法会在Rebalance之后调用
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                currentOffset.clear();
                for (TopicPartition partition : collection) {
                    //定位到最近提交的offset位置继续消费
                    consumer.seek(partition, getOffset(partition));
                }
            }
        });

        while (true) {
            //消费者拉取数据
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset=%d, key=%s, value=%s%n", record.offset(), record.key(), record.value());
                currentOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset());
            }
            commitOffset(currentOffset);//异步提交
        }

    }

    //提交该消费者所有分区的offset
    private void commitOffset(Map<TopicPartition, Long> currentOffset) {
    }

    //获取某分区的最新offset
    private long getOffset(TopicPartition partition) {
        return 0;
    }
}
